use rust_collector::exporter::queue::{ExporterQueue, QueueError};
use tokio::sync::mpsc;

#[derive(Clone, PartialEq, prost::Message)]
struct DummyMsg {
    #[prost(int32, tag = "1")]
    v: i32,
}

#[tokio::test]
async fn enqueue_and_worker_process_items() {
    let q = ExporterQueue::<DummyMsg>::new(8, None).await.unwrap();

    let (tx, mut rx) = mpsc::channel::<i32>(8);
    let ok = q
        .start_worker(move |msg: DummyMsg| {
            let tx = tx.clone();
            async move {
                let _ = tx.send(msg.v).await;
            }
        })
        .await;
    assert!(ok);

    q.enqueue(DummyMsg { v: 42 }).await.unwrap();
    q.enqueue(DummyMsg { v: 7 }).await.unwrap();

    let a = rx.recv().await.unwrap();
    let b = rx.recv().await.unwrap();
    assert_eq!((a, b), (42, 7));
}

#[tokio::test]
async fn graceful_shutdown_stops_worker_and_rejects_new_enqueues() {
    let q = ExporterQueue::<DummyMsg>::new(4, None).await.unwrap();
    let ok = q
        .start_worker(|_m| async move { /* drop */ })
        .await;
    assert!(ok);

    q.enqueue(DummyMsg { v: 1 }).await.unwrap();
    q.enqueue(DummyMsg { v: 2 }).await.unwrap();

    q.graceful_shutdown().await; // should not hang

    let res = q.enqueue(DummyMsg { v: 3 }).await;
    assert!(matches!(res, Err(QueueError::Closed)));
}

#[tokio::test]
async fn pause_and_resume_drops_queued_items_on_purge() {
    let q = ExporterQueue::<DummyMsg>::new(4, None).await.unwrap();

    let (tx, mut rx) = mpsc::channel::<i32>(4);
    q.start_worker(move |msg: DummyMsg| {
        let tx = tx.clone();
        async move {
            let _ = tx.send(msg.v).await;
        }
    })
    .await;

    // enqueue then purge
    q.enqueue(DummyMsg { v: 1 }).await.unwrap();
    q.enqueue(DummyMsg { v: 2 }).await.unwrap();
    q.pause(); // triggers purge of buffered items

    // resume and enqueue new value; only this should be delivered
    q.resume();
    q.enqueue(DummyMsg { v: 9 }).await.unwrap();

    let v = rx.recv().await.unwrap();
    assert_eq!(v, 9);
}


